﻿using System;
using System.Diagnostics;
using System.ServiceModel;
using System.Threading;
using Needle.Bus.Engine.Utils;
using Needle.Bus.Messages;
using SCG = System.Collections.Generic;

namespace Needle.Bus.Engine.Queues
{
    [DebuggerDisplay("BufferQueue {Id} contains {Count} {( Count == 1 ) ? \"message\" : \"messages\",nq}")]
    public class BufferQueue : EngineComponent, IBufferQueue
    {

        private readonly SCG.Queue<IMessage> _queue;
        private readonly AutoResetEvent _messagesEnqueued;

        #region constructors

        public BufferQueue(string id)
            : this(new NamedMutex(id), id)
        { }

        public BufferQueue(object mutex, string id)
            : base(mutex,id)
        {
            _queue = new SCG.Queue<IMessage>();
            _messagesEnqueued = new AutoResetEvent(false);
        }

        #endregion

        #region enqueue

        public void Enqueue(IMessage message)
        {
            lock (ThisLock)
            {
                if (!AcceptMessage(message))
                    throw new MessageRejectedException();
                _queue.Enqueue(message);
                _messagesEnqueued.Set();
            }
        }

        public void Enqueue(SCG.IList<IMessage> messages)
        {
            lock (ThisLock)
            {
                if (!AcceptMessages(messages))
                    throw new MessageRejectedException();
                foreach (var msg in messages)
                    _queue.Enqueue(msg);
                _messagesEnqueued.Set();
            }
        }

        public bool TryEnqueue(IMessage message)
        {
            lock (ThisLock)
            {
                if (AcceptMessage(message))
                {
                    _queue.Enqueue(message);
                    _messagesEnqueued.Set();
                    return true;
                }
                return false;
            }
        }

        public bool TryEnqueue(SCG.IList<IMessage> messages)
        {
            lock (ThisLock)
            {
                if (AcceptMessages(messages))
                {
                    foreach (var msg in messages)
                        _queue.Enqueue(msg);
                    _messagesEnqueued.Set();
                    return true;
                } 
                return false;
            }
        }

        public IDisposable PrepareForEnqueue(SCG.IList<IMessage> messages)
        {
            lock (ThisLock)
            {
                if (!AcceptMessages(messages))
                    throw new MessageRejectedException();
                return new SinkTransaction(this, messages);
            }
        }

        public void Commit(object transactionToken)
        {
            /* the transaction takes care of locking */
            var transaction = transactionToken as SinkTransaction;
            if (transaction == null || transaction.Sink != this)
                throw new ArgumentException("Invalid token", "transactionToken");
            transaction.Commit();
        }

        public void Abort(object transactionToken)
        {
            /* the transaction takes care of locking */
            var transaction = transactionToken as SinkTransaction;
            if (transaction == null || transaction.Sink != this)
                throw new ArgumentException("Invalid token", "transactionToken");
            transaction.Abort();
        }

        #endregion

        #region dequeue

        public bool TryDequeue(out IMessage message)
        {
            lock (ThisLock)
            {
                if (_queue.Count == 0)
                {
                    message = null;
                    return false;
                }
                message = _queue.Dequeue();
                return true;
            }
        }

        public bool TryDequeue(int maxCount, out SCG.IList<IMessage> messages)
        {
            messages = null;

            if (maxCount < 1)
                throw new ArgumentOutOfRangeException("maxCount must be larger than 0");

            lock (ThisLock)
            {
                int msgCount = Math.Min(_queue.Count, maxCount);

                if (msgCount == 0)
                    return false;

                var list = new SCG.List<IMessage>(msgCount);

                for (int i = 0; i < msgCount; i++)
                    list.Add(_queue.Dequeue());

                messages = list;
                return true;
            }
        }

        public IMessage Dequeue()
        {
            lock (ThisLock)
            {
                if (_queue.Count == 0)
                    throw new SourceEmptyException();
                return _queue.Dequeue();
            }
        }

        public IMessage Dequeue(TimeSpan timeOut)
        {
            Stopwatch sw = Stopwatch.StartNew();
            var timeLeft = timeOut;
            while (true)
            {
                lock (ThisLock)
                {
                    if (_queue.Count > 0)
                        return _queue.Dequeue();
                    else if (timeLeft <= TimeSpan.Zero)
                        throw new SourceEmptyException();
                }
                _messagesEnqueued.WaitOne(timeLeft, true);
                sw.Stop();
                timeLeft = timeOut - sw.Elapsed;
                sw.Start();
            }
        }

        public SCG.IList<IMessage> Dequeue(TimeSpan timeOut, int count)
        {
            Stopwatch sw = Stopwatch.StartNew();
            var timeLeft = timeOut;
            while (true)
            {
                lock (ThisLock)
                {
                    if (_queue.Count >= count)
                    {
                        SCG.List<IMessage> list = new SCG.List<IMessage>(count);
                        for (int i = 0; i < count; i++)
                            list.Add(_queue.Dequeue());
                        return list;
                    }
                    else if (timeLeft <= TimeSpan.Zero)
                    {
                        throw new SourceException("Insufficient number of messages");
                    }
                }
                _messagesEnqueued.WaitOne(timeLeft, true);
                sw.Stop();
                timeLeft = timeOut - sw.Elapsed;
                sw.Start();
            }
        }

        public SCG.IList<IMessage> Purge()
        {
            lock (ThisLock)
            {
                SCG.List<IMessage> list = new SCG.List<IMessage>(_queue.Count);
                while (_queue.Count > 0)
                    list.Add(_queue.Dequeue());
                return list;
            }
        }

        #endregion

        #region properties

        public int Count
        {
            get
            {
                lock (ThisLock)
                {
                    return _queue.Count;
                }
            }
        }

        private int _capacity = int.MaxValue;
        public int Capacity
        {
            get
            {
                lock (ThisLock)
                {
                    return _capacity;
                }
            }
            set
            {
                lock (ThisLock)
                {
                    if (value < 1)
                        _capacity = int.MaxValue;
                    else
                        _capacity = value;
                }
            }
        }

        private Predicate<IMessage> _filter;
        public Predicate<IMessage> Filter
        {
            get
            {
                lock (ThisLock)
                {
                    return _filter;
                }
            }
            set
            {
                lock (ThisLock)
                {
                    _filter = value;
                }
            }
        }

        #endregion

        #region privates

        private bool AcceptMessage(IMessage message)
        {
            if (message == null
                || IsDisposed
                || State != CommunicationState.Opened
                || _capacity < _queue.Count + 1)
                return false;

            var filter = _filter;
            return (filter == null || filter(message));
        }

        private bool AcceptMessages(SCG.IList<IMessage> messages)
        {
            if (messages == null
                || messages.Count == 0
                || IsDisposed
                || State != CommunicationState.Opened
                || _capacity < _queue.Count + messages.Count)
                return false;

            var filter = _filter;
            if (filter == null)
                return true;
            foreach (IMessage message in messages)
            {
                if (filter(message))
                    continue;
                else
                    return false;
            }
            return true;
        }

        #endregion

        protected override void OnOpen(TimeSpan timeout)
        {
            // TODO: implement
        }

        protected override void OnAbort()
        {
            // TODO: implement
        }

        protected override void OnClose(TimeSpan timeout)
        {
            // TODO: implement
        }
    }
}
